package com.xrui.hbase.schema.spec;

import com.google.common.base.Objects;
import com.google.common.base.Preconditions;

import com.xrui.hbase.AvroSchemaResolver;
import com.xrui.hbase.HBaseURI;
import com.xrui.hbase.codec.CellDecoderFactory;
import com.xrui.hbase.codec.CellEncoderFactory;
import com.xrui.hbase.codec.impl.DefaultCellDecoderFactory;
import com.xrui.hbase.codec.impl.DefaultCellEncoderFactory;
import com.xrui.hbase.exception.InvalidLayoutException;
import com.xrui.hbase.schema.CellSchema;
import com.xrui.hbase.schema.SchemaType;
import com.xrui.hbase.schema.impl.AvroCellSchema;
import org.apache.avro.Schema;
import org.apache.avro.specific.SpecificData;
import org.apache.avro.specific.SpecificRecord;
import org.apache.commons.lang.SerializationUtils;
import org.apache.hadoop.util.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Specification of a column in a HBase table.
 * <p>
 * </p>
 */
public final class CellSpec {
    private static final Logger LOG = LoggerFactory.getLogger(CellSpec.class);

    /**
     * HBaseURI of the column this specification is for. Potentially null.
     */
    private HBaseURI mColumnURI = null;

    /**
     * Avro record specifying the cell encoding.
     */
    private CellSchema mCellSchema = null;

    /**
     * Avro specific: decode the cell content with this Avro reader schema. Null otherwise.
     */
    private Schema mReaderSchema = null;

    /**
     * Avro specific: this is the default Avro reader schema, if any has been specified.
     * This field is lazily initialized as resolving the default reader schema requires
     * setting the schema table to use first.
     */
    private Schema mDefaultReaderSchema = null;

    /**
     * Schema resolver to resolve avro schema.
     */
    private AvroSchemaResolver mAvroSchemaResolver = null;

    /**
     * Factory for cell decoders (either specific or generic).
     */
    private CellDecoderFactory mDecoderFactory = DefaultCellDecoderFactory.get();

    /**
     * Factory for cell encoders.
     */
    private CellEncoderFactory mEncoderFactory = DefaultCellEncoderFactory.get();

    /**
     * Initializes a new unspecified CellSpec.
     */
    private CellSpec() {
    }

    /**
     * Constructs a CellSpec instance from a CellSchema descriptor.
     *
     * @param cellSchema  CellSchema record to initialize the CellSpec from.
     * @param avroSchemaResolver Schema resolver to resolve schema hashes or UIDs.
     * @return a new CellSpec initialized from a CellSchema descriptor.
     * @throws InvalidLayoutException if the cell specification is invalid.
     */
    public static CellSpec fromCellSchema(
        final CellSchema cellSchema,
        final AvroSchemaResolver avroSchemaResolver)
        throws InvalidLayoutException {
        return fromCellSchema(cellSchema)
            .setSchemaResolver(avroSchemaResolver);
    }

    /**
     * Constructs a CellSpec instance from a CellSchema descriptor.
     *
     * @param cellSchema CellSchema record to initialize the CellSpec from.
     * @return a new CellSpec initialized from the given CellSchema descriptor.
     * @throws InvalidLayoutException if the cell specification is invalid.
     */
    public static CellSpec fromCellSchema(final CellSchema cellSchema)
        throws InvalidLayoutException {
        return new CellSpec()
            .setCellSchema(cellSchema);
    }

    /**
     * Returns a new unspecified CellSpec.
     * <p>
     * <p> Use the setters to populate the necessary fields. </p>
     *
     * @return a new unspecified CellSpec.
     */
    public static CellSpec create() {
        return new CellSpec();
    }

    /**
     * Makes a copy of an existing CellSpec.
     *
     * @param spec Existing CellSpec to copy.
     * @return a copy of the specified CellSpec.
     */
    public static CellSpec copy(final CellSpec spec) {
        final CellSpec copy = new CellSpec();
        copy.mColumnURI = spec.mColumnURI;
        copy.mCellSchema = (CellSchema) SerializationUtils.clone(spec.getCellSchema());  // deep copy
        copy.mReaderSchema = spec.mReaderSchema;
        copy.mAvroSchemaResolver = spec.mAvroSchemaResolver;
        copy.mDecoderFactory = spec.mDecoderFactory;
        //copy.mEncoderFactory = spec.mEncoderFactory;
        return copy;
    }

    /**
     * Reports the Avro schema from a generated class.
     *
     * @param klass Generated class to report the schema of.
     * @return Avro schema of the specified generated class.
     * @throws InvalidLayoutException if the specified class does not correspond to a valid Avro type.
     */
    public static Schema getSchemaFromClass(Class<? extends SpecificRecord> klass)
        throws InvalidLayoutException {
        try {
            return SpecificData.get().getSchema(klass);
        } catch (RuntimeException re) {
            LOG.debug("Error accessing schema from Avro class: {}",
                StringUtils.stringifyException(re));
            throw new InvalidLayoutException("Java class is not a valid Avro type: " + klass.getName());
        }
    }

    /**
     * Sets the Avro schema to use when decoding cells.
     * <p>
     * <p> For Avro-encoded cells only. </p>
     *
     * @param readerSchema Avro reader schema.
     * @return this CellSpec.
     */
    public CellSpec setReaderSchema(final Schema readerSchema) {
        Preconditions.checkState(isAvro());
        mReaderSchema = Preconditions.checkNotNull(readerSchema);
        return this;
    }

    /**
     * Returns the schema resolver to resolve schema.
     *
     * @return the schema resolver to resolve schema.
     */
    public AvroSchemaResolver getSchemaResolver() {
        return mAvroSchemaResolver;
    }

    /**
     * Sets the Schema resolver used to resolve Avro schemas.
     *
     * @param avroSchemaResolver Schema resolver to resolve Avro schemas.
     * @return this CellSpec.
     */
    public CellSpec setSchemaResolver(final AvroSchemaResolver avroSchemaResolver) {
        mAvroSchemaResolver = avroSchemaResolver;
        return this;
    }

    /**
     * Returns the factory for cell decoders to use for this column.
     *
     * @return the factory for cell decoders to use for this column.
     */
    public CellDecoderFactory getDecoderFactory() {
        return mDecoderFactory;
    }

    /**
     * Sets the factory for cell decoders (either specific or generic).
     *
     * @param decoderFactory Factory for cell decoders (either specific or generic).
     * @return this CellSpec.
     */
    public CellSpec setDecoderFactory(final CellDecoderFactory decoderFactory) {
        mDecoderFactory = decoderFactory;
        return this;
    }

    /**
     * Returns the factory for cell encoders to use for this column.
     *
     * @return the factory for cell encoders to use for this column.
     */
    public CellEncoderFactory getEncoderFactory() {
        return mEncoderFactory;
    }

    /**
     * Sets the factory for cell encoders.
     *
     * @param encoderFactory Factory for cell encoders.
     * @return this CellSpec.
     */
    public CellSpec setEncoderFactory(final CellEncoderFactory encoderFactory) {
        mEncoderFactory = encoderFactory;
        return this;
    }

    /**
     * Returns whether this cell is encoded with Avro.
     *
     * @return whether this cell is encoded with Avro.
     */
    public boolean isAvro() {
        Preconditions.checkNotNull(mCellSchema);
        return (mCellSchema.getType() == SchemaType.AVRO);
    }

    /**
     * Returns whether this cell is encoded with Protobuf.
     *
     * @return whether this cell is encoded with Protobuf.
     */
    public boolean isProtobuf() {
        Preconditions.checkNotNull(mCellSchema);
        return (mCellSchema.getType() == SchemaType.PROTOBUF);
    }

    /**
     * Returns the underlying CellSchema Avro record. May be null.
     *
     * @return the underlying CellSchema Avro record. May be null.
     */
    public CellSchema getCellSchema() {
        return mCellSchema;
    }

    /**
     * Initializes this CellSpec instance from a given CellSchema specification.
     *
     * @param cellSchema CellSchema to initialize the CellSpec from.
     * @return this CellSpec.
     * @throws InvalidLayoutException on layout error.
     */
    public CellSpec setCellSchema(final CellSchema cellSchema) throws InvalidLayoutException {
        mCellSchema = (CellSchema) SerializationUtils.clone(cellSchema);  // deep copy
        if (isAvro()) {
            String json = ((AvroCellSchema)cellSchema).getSchemaJson();
            try {
                mReaderSchema = new Schema.Parser().parse(json);
            } catch (RuntimeException re) {
                throw new InvalidLayoutException(String.format(
                    "Invalid Avro schema '%s' caused error: %s.",
                    json, re.getMessage()));
            }
        }
        return this;
    }

    /**
     * Returns the Avro reader schema used to decode cells.
     * <p>
     * <p> For Avro-encoded cells only. </p>
     *
     * @return the Avro reader schema used to decode cells.
     * Null means cells are decoded using their writer schema.
     */
    public Schema getAvroSchema() {
        Preconditions.checkState(isAvro());
        return mReaderSchema;
    }

    /**
     * Reads the Avro schema from the table layout.
     *
     * @param cellSchema The portion of the table layout record to read from.
     * @return the Avro schema if relevant, or null.
     * @throws InvalidLayoutException if the specification or the schema is invalid.
     */
    public static Schema readAvroSchema(final CellSchema cellSchema) throws InvalidLayoutException {
        switch (cellSchema.getType()) {
            case AVRO: {
                try {
                    return new Schema.Parser().parse(cellSchema.getValue());
                } catch (RuntimeException re) {
                    throw new InvalidLayoutException(String.format(
                        "Invalid Avro schema: '%s' caused error: %s.",
                        cellSchema.getValue(), re.getMessage()));
                }
            }
            case PROTOBUF: {
                // Protocol-buffer have no Avro schema.
                return null;
            }
            case RAW_BYTES: {
                // No Avro schema for raw-byte cells:
                return null;
            }
            case CUSTOM: {
                // No Avro schema for custom type cells:
                return null;
            }
            case CDFSTORE: {
                // No Avro schema for custom type cells:
                return null;
            }
            default:
                throw new InvalidLayoutException("Invalid schema type: " + cellSchema.getType());
        }
    }

    /**
     * Returns the HBaseURI of the column this specification is for. Potentially null.
     *
     * @return the HBaseURI of the column this specification is for. Potentially null.
     */
    public HBaseURI getColumnURI() {
        return mColumnURI;
    }

    /**
     * Sets the HBaseURI of the column this specification is for.
     *
     * @param uri HBaseURI of the column this specification is for. May be null.
     * @return this CellSpec.
     */
    public CellSpec setColumnURI(HBaseURI uri) {
        Preconditions.checkArgument((uri == null) || (uri.getColumnNames().size() == 1),
            "Invalid CellSpec HBaseURI '%s' : CellSpec requires a HBaseURI for a single column.", uri);
        mColumnURI = uri;
        return this;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public String toString() {
        return Objects.toStringHelper(CellSpec.class)
            .add("column_uri", mColumnURI)
            .add("cell_schema", mCellSchema)
            .add("reader_schema", mReaderSchema)
            .toString();
    }

}
